package com.shujia.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo2SourceFunction {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 自定义source
      * 实现SourceFunction接口
      * 实现里面的run方法----> 在run方法中获取数据发送到下游
      *
      *
      */

    val ds: DataStream[Int] = env.addSource(new MySourceFunction)

    ds.print()

    env.execute()

  }

}

class MySourceFunction extends SourceFunction[Int]{

  /**
    * 在run 方法中获取数据，将数据发送到下游
    *
    * 可以在run  方法中链接外部的数据源获取数据
    *
    * @param ctx flink source的上下文对象，用于将数据发送到哦下游
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    var i = 0
    while (i < 10){

      //将数据发送到下游
      ctx.collect(i)

      i+=1
      Thread.sleep(1000)
    }

  }

  /**
    * 任务在取消的时候调用，用于回收资源（关闭jdbc 链接）
    *
    */

  override def cancel(): Unit = ???
}
